/*
    主题发布订阅业务层客户端：分为两种情况，一种是：主题发布客户端，另一种是：主题订阅客户端，但是两个有公共的部分：主题的创建/删除；
        发布客户端：可以主题的创建/删除/消息发布
        订阅客户端：
            1、可以主题的创建/删除/订阅/取消订阅
            2、提供接收消息发布的回调函数，被注册给Dispatcher模块
            3、需要维护一个关系表<主题，回调函数>，因为对于主题的订阅，当订阅完成的时候，不一定能立马接收到消息的发布，所以消息的订阅一定是一个异步请求，
               不然会阻塞主线程，采用异步回调的方式来处理，当接收到消息发布的请求时，根据不同的主题来调用对应的回调
    由于两者有相似性，所以实现为一个模块，即TopicManager，最后将网络模块融合封装客户端的时候可以分来分装
*/
#pragma once
#include "../common/message.hpp"
#include "../common/uuid.hpp"
#include "requestor.hpp"

namespace ns_jsonrpc
{
    namespace ns_client
    {
        // 主题管理
        class TopicManager
        {
        public:
            using TopicManagerPtr = std::shared_ptr<TopicManager>;
            using SubscribeCallback = std::function<void(const std::string &, const std::string &)>;

        public:
            TopicManager(const Requestor::RequestorPtr &requestor) : _requestor(requestor)
            {
            }

            // 主题创建
            bool topicCreate(const std::string &topicName, const BaseConnection::BaseConnectionPtr &conn)
            {
                bool ret = topicRequestHandler(topicName, conn, TopicOpType::TOPIC_CREATE);
                if (!ret)
                {
                    LOG_ERROR("topic create failed, topic name is %s\n", topicName.c_str());
                    return false;
                }
                LOG_INFO("topic create success, topic name is %s\n", topicName.c_str());
                return true;
            }

            // 主题删除
            bool topicRemove(const std::string &topicName, const BaseConnection::BaseConnectionPtr &conn)
            {
                bool ret = topicRequestHandler(topicName, conn, TopicOpType::TOPIC_REMOVE);
                if (!ret)
                {
                    LOG_ERROR("topic remove failed, topic name is %s\n", topicName.c_str());
                    return false;
                }
                LOG_INFO("topic remove success, topic name is %s\n", topicName.c_str());
                return true;
            }

            // 主题订阅
            bool topicSubscribe(const std::string &topicName, const BaseConnection::BaseConnectionPtr &conn, const SubscribeCallback &subscribeCallback)
            {
                // 应该先设置回调，再订阅，因为有可能订阅完就直接能接收到消息的转发，而此时如果回调还没有设置好，就会出现问题
                // 同时，如果订阅失败，则也要删除回调
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _topics[topicName] = subscribeCallback;
                }
                bool ret = topicRequestHandler(topicName, conn, TopicOpType::TOPIC_SUBSCRIBE);
                if (!ret)
                {
                    // 订阅失败，则要将回调删除
                    std::unique_lock<std::mutex> lock(_mutex);
                    _topics.erase(topicName);
                    LOG_ERROR("topic subscribe failed, topic name is %s\n", topicName.c_str());
                    return false;
                }
                LOG_INFO("topic subscribe success, topic name is %s\n", topicName.c_str());
                return true;
            }

            // 主题取消订阅
            bool topicCancel(const std::string &topicName, const BaseConnection::BaseConnectionPtr &conn)
            {
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _topics.erase(topicName);
                }
                bool ret = topicRequestHandler(topicName, conn, TopicOpType::TOPIC_CANCEL);
                if (!ret)
                {
                    LOG_ERROR("topic cancel failed, topic name is %s\n", topicName.c_str());
                    return false;
                }
                LOG_INFO("topic cancel success, topic name is %s\n", topicName.c_str());
                return true;
            }

            // 消息发布
            bool topicPublish(const std::string &topicName, const BaseConnection::BaseConnectionPtr &conn, const std::string &msgPublish)
            {
                bool ret = topicRequestHandler(topicName, conn, TopicOpType::TOPIC_PUBLISH, msgPublish);
                if (!ret)
                {
                    LOG_ERROR("topic public failed, topic name is %s\n", topicName.c_str());
                    return false;
                }
                LOG_INFO("topic public success, topic name is %s\n", topicName.c_str());
                return true;
            }

            // 接收消息发布请求的回调，被注册给Dispatcher模块
            void onTopicPublish(const BaseConnection::BaseConnectionPtr &conn, const TopicRequest::TopicRequestPtr &msg)
            {
                TopicOpType optype = msg->opType();
                if (optype != TopicOpType::TOPIC_PUBLISH)
                {
                    LOG_ERROR("topic publish handle error, invalid optype\n");
                    return;
                }
                auto it = _topics.find(msg->key());
                if (it == _topics.end())
                {
                    LOG_ERROR("topic publish handle error, subscribe callback not found\n");
                    return;
                }
                SubscribeCallback cb = it->second;
                cb(msg->key(), msg->msg());
            }

        private:
            bool topicRequestHandler(const std::string topicName, const BaseConnection::BaseConnectionPtr &conn, TopicOpType optype, const std::string &msgPublish = "")
            {
                // 1、构建请求
                TopicRequest::TopicRequestPtr req = MessageFactory::create<TopicRequest>();
                req->setId(UUID::uuid());
                req->setMType(MType::REQ_TOPIC);
                req->setKey(topicName);
                req->setOpType(optype);
                if (optype == TopicOpType::TOPIC_PUBLISH)
                    req->setMsg(msgPublish);

                // 2、发送请求 & 获取响应
                BaseMessage::BaseMessagePtr rsp;
                _requestor->send(conn, req, rsp);

                // 3、处理响应
                TopicResponse::TopicResponsePtr rspTopic = std::dynamic_pointer_cast<TopicResponse>(rsp);
                if (rspTopic == nullptr || rspTopic->check() == false)
                {
                    LOG_ERROR("topic request is error, invalid response\n");
                    return false;
                }
                if (rspTopic->rcode() != RCode::RCODE_OK)
                {
                    LOG_ERROR("topic request is error, error reason: %s\n", errReason(rspTopic->rcode()));
                    return false;
                }
                return true;
            }

        private:
            Requestor::RequestorPtr _requestor;
            std::unordered_map<std::string, SubscribeCallback> _topics;
            std::mutex _mutex;
        };
    }
}